package com.ptone.messagecenter.modules.msg.kafka.controller;

import com.alibaba.fastjson.JSONObject;
import com.ptone.messagecenter.common.base.Result;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.*;

import javax.annotation.Resource;

@RestController
@RequestMapping("/producer")
@Slf4j
public class ProducerController {

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    @PostMapping("/produceWithOutPartition")
    public Result<Boolean> produce(@RequestParam("topicName") String topicName, @RequestParam("messageKey") String messageKey, @RequestBody JSONObject jsonObject) {
        String messageValue = JSONObject.toJSONString(jsonObject);
        kafkaTemplate.send(topicName, messageKey, messageValue);
        log.trace("produceWithOutPartition接口，messageValue：{}", messageValue);
        return Result.success(true);
    }

    @PostMapping("/produce")
    public Result<Boolean> produce(@RequestParam("topicName") String topicName, @RequestParam("partition") int partition, @RequestParam("messageKey") String messageKey, @RequestParam("messageValue") String messageValue) {
        kafkaTemplate.send(topicName, partition, messageKey, messageValue);
        log.trace("produce接口，topicName：{}", topicName);
        log.trace("produce接口，partition：{}", partition);
        log.trace("produce接口，messageKey：{}", messageKey);
        return Result.success(true);
    }
}
